package net.gz01.wallet.web.noitfy;



import net.gz01.wallet.web.model.TradeOrder;

import java.util.ArrayList;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.*;
import java.util.concurrent.locks.Lock;


/**
 * 推送分发
 * @author zhongzhanfeng
 *
 */
public class NotifyThread {

	/**
	 * 每一个线程处理的数量
	 */
	private static final int PRO_NUMBER = 20;
	private static ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
	private static Vector<NotifyThreadData> notifyList = new Vector<>();
	private static boolean run = true;

	private static final NotifyThread INSTANCE = new NotifyThread();

	private static Thread whileThread;
	private static boolean isNotify = false;

	private NotifyThread(){

	}

	public static final NotifyThread getInstance(){
		return NotifyThread.INSTANCE;
	}

	public void stop(){
		System.out.println("------------------- 正在关闭通知商户线程池 -------------------");
		run = false;
		try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
        	System.out.println("------------------- 通知商户线程池关闭异常 -------------------\r\n" + e.getMessage());
        }
		cachedThreadPool.shutdown();
		if(cachedThreadPool.isShutdown()) {
			System.out.println("cachedThreadPool is shutdown");
		}
		whileThread.interrupt();
		System.out.println("------------------- 通知商户线程池关闭成功 -------------------");
	}

	public void start(){
		whileThread = new Thread(new Runnable() {
			@Override
			public void run() {
				NotifyThread thread = new NotifyThread();
				thread.run();
			}
		});
		whileThread.start();
		System.out.println("------------------- 通知商户线程池启动成功 -------------------");
	}

	public void addNotifyOrder(TradeOrder order) {
		addNotifyOrder(new NotifyReq(order));
	}

	/**
	 * 明确失败或成功才会推送给商户
	 * @param req
	 */
	public void addNotifyOrder(NotifyReq req){
		synchronized (Lock.class) {
			if(isNotify == false) {
				Lock.class.notify();
			}
        }
		notifyList.add(new NotifyThreadData(req));
	}

	public void run(){
		CompletionService<List<NotifyThreadData>> pool = new ExecutorCompletionService<List<NotifyThreadData>>(cachedThreadPool);
		try{

			int size;
			while(run){

				size = notifyList.size();
				if(size == 0){
					//无数据时让线程阻塞
					synchronized (Lock.class) {
						try {
							Lock.class.wait();
							isNotify = false;
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				}


				List<Vector<NotifyThreadData>> que = new ArrayList<Vector<NotifyThreadData>>();
				Vector<NotifyThreadData> temp = new Vector<NotifyThreadData>();

				//将总集合数据根据20条拆分成多个集合
				for (int i = 0; i < size; i++) {
					temp.add(notifyList.get(i));
					if((i + 1) % PRO_NUMBER == 0){
						que.add(temp);
						temp.clear();
					}
				}

				if(temp.size() > 0){
					que.add(temp);
				}
				for (Vector<NotifyThreadData> notifyProcess : que) {
					pool.submit(new NotifyProcessThread(notifyProcess));
				}

				//线程执行结果
				for(int i = 0; i < que.size() ; i++){
				   List<NotifyThreadData> removeList = pool.take().get();
				   notifyList.removeAll(removeList);
				}

				Thread.sleep(3000);
			}

		}catch(InterruptedException e){
			System.out.println("获取子线程操作结果发生异常->"+e.getMessage());
		} catch (ExecutionException e) {
			System.out.println("线程池发生异常->"+e.getMessage());
		}
	}

}
